package com.qwiki.test;

import java.io.IOException;

import com.qwiki.xml.WikiPageAbstractAsyncHandler;
import com.qwiki.xml.WikiPageAsyncWordCountHandler;
import com.qwiki.xml.WikiPageAsyncWordDocFreqHandler;
import com.qwiki.xml.WikiPageHandler;
import com.qwiki.xml.WikiPageMultiHandler;
import com.qwiki.xml.WikiXMLPageValidArticleFilter;
import com.qwiki.xml.WikiXmlEngine;

import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;
import org.xml.sax.SAXException;

import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
import it.unimi.dsi.fastutil.objects.ObjectIterator;

public class wordsAnalysis {
  public static void main(String[] args) throws SAXException, IOException, InterruptedException {
    Logger.getRootLogger()
        .addAppender(new ConsoleAppender(new PatternLayout("%d{yyyy/MM/dd HH:mm:ss.SSS} %p [%c{1}] [%x] %m%n")));

    WikiXmlEngine engine = new WikiXmlEngine(args[0]);
    int numThreads = Integer.parseInt(args[1]);

    WikiPageAbstractAsyncHandler[] handlers = new WikiPageAbstractAsyncHandler[numThreads];
    if (args.length == 3 && args[2].equalsIgnoreCase("docfreq")) {
      for (int i = 0; i < handlers.length; i++) {
        handlers[i] = new WikiPageAsyncWordDocFreqHandler("test", i);
        handlers[i].start();
      }
    } else {
      for (int i = 0; i < handlers.length; i++) {
        handlers[i] = new WikiPageAsyncWordCountHandler("test", i);
        handlers[i].start();
      }
    }
    WikiPageHandler handler = new WikiPageMultiHandler(handlers);
    handler.setFilter(new WikiXMLPageValidArticleFilter());
    engine.setHandler(handler);
    engine.process();
    Object2IntOpenHashMap<String> counters = new Object2IntOpenHashMap<String>();
    counters.defaultReturnValue(0);
    for (int i = 0; i < handlers.length; i++) {
      handlers[i].drain();
      System.out.println("thread " + i + " drained");
      handlers[i].join();
      merge(counters, getCounter(handlers[i]));
    }
    // print out
    ObjectIterator<String> keys = counters.keySet().iterator();
    while (keys.hasNext()) {
      String key = keys.next();
      int count = counters.getInt(key);
      System.out.println("Wordcount \t" + count + "\t" + ":" + key);
    }
  }

  private static Object2IntOpenHashMap<String> getCounter(WikiPageAbstractAsyncHandler handler) {
    if (handler instanceof WikiPageAsyncWordCountHandler) {
      return ((WikiPageAsyncWordCountHandler) handler).counters;
    } else if (handler instanceof WikiPageAsyncWordDocFreqHandler) {
      return ((WikiPageAsyncWordDocFreqHandler) handler).counters;
    }
    throw new RuntimeException("bad handler type " + handler.getClass());
  }

  private static Object2IntOpenHashMap<String> merge(Object2IntOpenHashMap<String> counters1,
                                                     Object2IntOpenHashMap<String> counters2) {
    ObjectIterator<String> keys = counters2.keySet().iterator();
    while (keys.hasNext()) {
      String key = keys.next();
      counters1.put(key, counters1.getInt(key) + counters2.getInt(key));
    }
    return counters1;
  }
}
